package com.mjf.market_analysis

import java.sql.Timestamp

import com.mjf.dim.{MarketCount, MarketUserBehavior}
import com.mjf.source.SimulateMarketEventSource
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 分渠道APP市场统计
 */
object AppMarketingByChannel {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val dataStream: DataStream[MarketUserBehavior] = env.addSource(new SimulateMarketEventSource)
      .assignAscendingTimestamps(_.timestamp)

    val resultStream: DataStream[MarketCount] = dataStream
      .filter(_.behavior != "UNINSTALL") // 过滤到卸载行为
      .keyBy(data => (data.channel, data.behavior)) // 按照渠道和行为类型分组
      .timeWindow(Time.hours(1), Time.seconds(5))
      .process(new MarketCountByChannel())  // 自定义全窗口函数

    resultStream.print()

    env.execute("AppMarketingByChannel")

  }
}

// 自定义ProcessWindowFunction
class MarketCountByChannel() extends ProcessWindowFunction[MarketUserBehavior, MarketCount, (String, String), TimeWindow] {
  override def process(key: (String, String), context: Context, elements: Iterable[MarketUserBehavior], out: Collector[MarketCount]): Unit = {

    val windowStart: String = new Timestamp(context.window.getStart).toString
    val windowEnd: String = new Timestamp(context.window.getEnd).toString
    val channel: String = key._1
    val behavior: String = key._2
    val count: Long = elements.size

    out.collect(MarketCount(windowStart, windowEnd, channel, behavior, count))

  }
}